package cn.itcast.tags.tools


import org.apache.spark.ml.linalg
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.ml.clustering.KMeansModel
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 *针对标签进行相关操作工具类
 */
object TagTools {

  /**
   * KMeans 算法标签模型基于属性规则标签数据和KMeansModel预测值打标签封装方法
   * @param kmeansModel KMeans 算法模型
   * @param dataframe 预测数据集，含预测列prediction
   * @param tagDF 标签数据
   * @return 用户标签数据，含uid和tagId列
   */
  def kmeansMatchTag(kmeansModel: KMeansModel, dataframe: DataFrame,
                     tagDF: DataFrame,tagColumn:String): DataFrame = {
    val spark: SparkSession = dataframe.sparkSession
    import spark.implicits._
    // 1. 获取聚类模型中簇中心及索引
    val clusterCenters: Array[linalg.Vector] = kmeansModel.clusterCenters
    // 2. 类簇中心点对应标签ID
    val indexTagMap: Map[Int, String] = TagTools.convertIndexMap(clusterCenters, tagDF)
    val indexTagMapBroadcast = spark.sparkContext.broadcast(indexTagMap)
    // 3. 关联聚类预测值，将prediction转换为tagId
    val index_to_tag = udf(
      (prediction: Int) => indexTagMapBroadcast.value(prediction)
    )
    dataframe.select(
      $"userId", //
      index_to_tag($"prediction").as(tagColumn) //
    )
  }


  /**
   * 将KMeans模型中类簇中心点索引对应到属性标签的标签ID
   * @param clusterCenters KMeans模型类簇中心点
   * @param tagDF 属性标签数据
   */
  def convertIndexMap(clusterCenters: Array[linalg.Vector],
                      tagDF: DataFrame): Map[Int, String] = {

    //列簇中心点坐标
    val centerIndexArray: Array[((Int, Double), Int)] = clusterCenters
      .zipWithIndex
      .map{
        case (clusterVector,clusterIndex)=>
          //rfm表示R+F+M,越大表示客户价值越高
          val rfm = clusterVector.toArray.sum//对每个列簇的向量累加
          clusterIndex->rfm
      }
      .sortBy(tuple=> - tuple._2) //降序
      .zipWithIndex
    //((2,12.0),0)

    //打标签
    val ruleMap: Map[String, String] = TagTools.convertMap(tagDF)
    ruleMap.foreach(println) //(4,低价值)

    val indexTagMap: Map[Int, String] = centerIndexArray
      .map {
        case ((centerIndex, _), index) =>
          val tagName = ruleMap(index.toString)
          (centerIndex, tagName)
      }.toMap
    indexTagMap
  }


  /**
   * 将标签数据中属性标签规则rule拆分为范围: start, end
   * @param tagDF 标签数据
   * @return 数据集DataFrame
   */
  def convertTuple(tagDF: DataFrame): DataFrame = {
    // 导入隐式转换和函数库
    import tagDF.sparkSession.implicits._
    import org.apache.spark.sql.functions._
    // 1. 自定UDF函数，解析分解属性标签的规则rule： 19500101-19591231
    val rule_to_tuple: UserDefinedFunction = udf(
      (rule: String) => {
        val Array(start, end) = rule.split("-").map(_.toInt)
        // 返回二元组
        (start, end)
      }
    )
    // 2. 获取属性标签数据，解析规则rule
    val ruleDF: DataFrame = tagDF
      .filter($"level" === 5) // 5级标签
      .select(
        $"name", //
        rule_to_tuple($"rule").as("rules") //
      )
      // 获取起始start和结束end
      .select(
        $"name", //
        $"rules._1".as("start"), //
        $"rules._2".as("end") //
      )
    //ruleDF.show(20, truncate = false)
    // 3. 返回标签规则
    ruleDF
  }

  /**
   *将[属性标签]数据中[规则：rule与名称：name]转换为[Map集合]
   * @param tagDF 属性标签数据 * @return Map 集合
   */
  def convertMap(tagDF: DataFrame): Map[String, String] = {
    import tagDF.sparkSession.implicits._
    tagDF
      // 获取属性标签数据
      .filter($"level" === 5)
      // 选择标签规则rule和标签Id
      .select($"rule", $"name")
      // 转换为Dataset
      .as[(String, String)]
      // 转换为RDD
      .rdd
      // 转换为Map集合
      .collectAsMap().toMap
  }

  /**
   * 依据标签业务字段值与标签规则进行匹配，进行打标签
   * @param dataFrame：业务数据
   * @param field ：业务字段
   * @param tagDF ：标签数据
   * @return ：标签模型数据
   */
  def ruleMatchTag(dataFrame: DataFrame,field:String,tagDF:DataFrame): DataFrame ={
    val spark: SparkSession = dataFrame.sparkSession
    import spark.implicits._

    //1.获取规则与tagId集合
    val attrTagRuleMap: Map[String, String] = convertMap(tagDF) //rule->name

    //2、将Map集合数据广播出去
    val attrTagRuleMapBroadcast: Broadcast[Map[String, String]] = spark.sparkContext.broadcast(attrTagRuleMap)

    //3.自定义UDF函数，依据job职业和属性标签规则进行标签化
    val field_to_tag = udf(
      (field:String) => attrTagRuleMapBroadcast.value(field)
    )

    //4.计算标签，依据业务字段值获取标签ID
    val modelDF: DataFrame = dataFrame.select(
      $"id".as("userId").cast(StringType),
      field_to_tag(col(field)).as(field)
    )
    //5.返回模型
    modelDF
  }



}
